package spike.pool;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by dmitry on 10/12/14.
 */
public class MyConnectionPool<T> implements ConnectionPool<T> {

    private final ConnectionSource<T> connectionSource;
    private final int capacity;
    private final int initNumber;
    private final long keepAliveTime;
    private final TimeUnit timeUnit;
    private boolean isStopped = false;

    private AtomicInteger openedConnectionsNumber = new AtomicInteger(0);
    private final ConcurrentLinkedDeque<T> freeConnections = new ConcurrentLinkedDeque<>();


    public MyConnectionPool(int capacity, ConnectionSource<T> connectionSource) {
        this.connectionSource = connectionSource;
        this.capacity = capacity;
        this.initNumber = capacity;
        this.keepAliveTime = 0;
        this.timeUnit = TimeUnit.NANOSECONDS;
    }


    public MyConnectionPool(int initNumber, int maxNumber, long keepAliveTime, TimeUnit timeUnit, ConnectionSource<T> connectionSource) {
        this.connectionSource = connectionSource;
        this.capacity = maxNumber;
        this.initNumber = initNumber;
        this.keepAliveTime = keepAliveTime;
        this.timeUnit = timeUnit;

    }


    // a procedure to prepare initNumber of already opened connections
    public void start() throws ExecutionException, InterruptedException {

        isStopped = false;

        LinkedList<Future<T>> localTasks = new LinkedList<>();

        for (int i = 0; i < initNumber; i++) {
            FutureTask<T> ft = new FutureTask<T>(() -> {
               return connectionSource.open();
            });

            ft.run();

            localTasks.add(ft);
        }

        for (Future<T> f: localTasks) {
            freeConnections.addLast(f.get());
        }

        openedConnectionsNumber.set(initNumber);

        this.backgroundCloseConnections();
    }


    // Mark as stopped. Stops after all connections released
    public void stop() {
        isStopped = true;
    }

    private void backgroundCloseConnections() {
        Thread td = new Thread(() -> {

            while (!isStopped) {

                try {
                    Thread.currentThread().sleep(TimeUnit.MILLISECONDS.convert(keepAliveTime, timeUnit));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                if (freeConnections.size() > initNumber) {
                    T connection = freeConnections.pollFirst();
                    openedConnectionsNumber.decrementAndGet();
                    synchronized (openedConnectionsNumber) {
                        openedConnectionsNumber.notify();
                    }
                    connectionSource.close(connection);
                }
                else {
                    try {
                        synchronized (freeConnections) {
                            freeConnections.wait();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }

            // stopped. Close connections until all are closed

            while (openedConnectionsNumber.get() > 0) {
                if (!freeConnections.isEmpty()) {
                    T connection = freeConnections.pollFirst();
                    openedConnectionsNumber.decrementAndGet();
                    connectionSource.close(connection);
                }
                else {
                    try {
                        synchronized (freeConnections) {
                            freeConnections.wait();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        td.start();
    }


    @Override
    public T get() throws InterruptedException {

        if (isStopped) {
            return null;
        }

        T connection = freeConnections.pollLast();

        if (connection == null) {

            while(true) {

                if (openedConnectionsNumber.getAndIncrement() < capacity) {

                    FutureTask<T> ft = new FutureTask<T>(() -> {
                        return connectionSource.open();
                    });

                    ft.run();

                    Future<T> f = ft;

                    try {
                        return f.get();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                } else {
                    openedConnectionsNumber.decrementAndGet();
                    synchronized (openedConnectionsNumber) {
                        openedConnectionsNumber.wait();
                    }
                    continue;
                }
            }
        }
        else {
            return connection;
        }

    }

    @Override
    public T tryGet() {
        return freeConnections.pollLast();
    }

    @Override
    public T tryGet(long time, TimeUnit unit) throws InterruptedException {
        T connection = freeConnections.pollLast();

        if (connection != null) {
            return connection;
        }

        synchronized (freeConnections) {
            freeConnections.wait(TimeUnit.MILLISECONDS.convert(keepAliveTime, timeUnit));
        }

        connection = freeConnections.pollLast();

        return connection;
    }

    @Override
    public void release(T object) {

        freeConnections.addLast(object);

        synchronized (freeConnections) {
            freeConnections.notify();
        }
    }
}
